fix(metrics): correct GFE metrics extraction and enable by default - #17561
fix(metrics): correct GFE metrics extraction and enable by default#17561sinhasubham wants to merge 6 commits into
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces deferred metrics recording for streaming and async streaming RPC responses in Google Cloud Spanner by wrapping responses in specialized wrappers. It also implements GFE latency extraction from response metadata and adds corresponding tests. The review feedback highlights several critical improvements for robustness: refining streaming response detection to check for iterator methods (next and anext) rather than iterable methods to prevent incorrect wrapping of standard iterables; wrapping telemetry and metrics recording blocks in try-except blocks to ensure telemetry failures do not disrupt the main application flow; defensively validating metadata elements before unpacking to avoid unpacking errors; and properly decoding bytes metadata values before regex matching.
|
@sinhasubham, Please merge main into this branch. The presubmit failures were addressed in #17578 |
a495bbb to
82f08c2
Compare
6c8800b to
dd8522b
Compare
| instrument_gfe_latency: Optional["Histogram"] = None, | ||
| instrument_gfe_missing_header_count: Optional["Counter"] = None, | ||
| instrument_afe_latency: Optional["Histogram"] = None, | ||
| instrument_afe_missing_header_count: Optional["Counter"] = None, |
| self._tracer.record_gfe_metrics(metadata) | ||
| self._tracer.record_afe_metrics(metadata) |
There was a problem hiding this comment.
Both methods are extracting latency values from metadata seprarately, this can be optimized to fetch both afe and gfe latency values from the metadata once, and pass it in these recorder methods
| if hasattr(response, "__next__"): | ||
| return _StreamingResponseWrapper(response, tracer) |
There was a problem hiding this comment.
Instead of checking this, cant we have interceptor listening to different events.
For eg: refer Node code, similar thing is used in Java and Go as well https://github.com/googleapis/google-cloud-node/blob/main/handwritten/spanner/src/metrics/interceptor.ts#L65
We have events like onReceiveStatus onReceiveMetadata ,
onReceiveStatus will only be called after the requests has finished execution which should trigger record_attempt_completion
onReceiveMetadata should be the place where we capture the metadata and use the value while calling record_gfe_metrics from onReceiveStatus
There was a problem hiding this comment.
Unlike Node.js, Java or Go: Python grpc and grpc.aio client interceptor APIs do not natively expose event-based lifecycle hooks (like onReceiveMetadata or onReceiveStatus).
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request adds support for collecting and exporting Application Front End (AFE) metrics alongside Google Front End (GFE) metrics, and introduces an AsyncMetricsInterceptor to handle asynchronous gRPC calls. The review feedback highlights critical issues: first, passing call_details.metadata directly to _extract_resource_from_path in the async interceptor will raise an AttributeError because it is a grpc.aio.Metadata object rather than a dictionary; second, metadata keys can be bytes in some gRPC environments, meaning str(key) will produce "b'server-timing'" and fail string comparisons in both extract_gfe_latency and extract_afe_latency.
| METRIC_NAME_AFE_LATENCY = "afe_latency" | ||
| METRIC_NAME_AFE_MISSING_HEADER_COUNT = "afe_missing_header_count" |
There was a problem hiding this comment.
Metric names are "gfe_latencies" and "afe_latencies" .
Not sure how were you able to export metrics to Cloud Monarch and Monitoring ? The screenshot attached shows metrics in "Generic Task" , it should be under Spanner -> Client with full name as spanner.googleapis.com/client/afe_latencies
There was a problem hiding this comment.
Addressed. updated test screenshot
ceaa773 to
644fe1c
Compare
644fe1c to
df80428
Compare
surbhigarg92
left a comment
There was a problem hiding this comment.
If possible also add a integration test similar to https://github.com/googleapis/google-cloud-java/blob/main/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBuiltInMetricsTest.java#L87
| ) | ||
| method_str = call_details.method | ||
| if isinstance(method_str, bytes): | ||
| method_str = method_str.decode("utf-8") |
There was a problem hiding this comment.
Why do we need this ? Why would method name have the encoding ? It was not there earlier, which case was failing because of it ?
There was a problem hiding this comment.
This is needed for async flow as in the asynchronous grpc.aio library, the underlying C-extension passes call_details.method as a byte string. So this is needed in async_intercept but not mandatory for sync intercept. So i have removed this from sync intercept and kept utf-8 decoding in async_intercept.
| tracer.set_method(method_name) | ||
| tracer.record_attempt_start() | ||
|
|
||
| if os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true": |
There was a problem hiding this comment.
Instead of getting this from ENV for every call, can we fetch it once during client init and pass it to the interceptor ?
| if os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true": | ||
| metadata = list(call_details.metadata or []) | ||
| metadata.append(("x-goog-spanner-enable-afe-server-timing", "true")) | ||
| call_details = call_details._replace(metadata=metadata) | ||
|
|
There was a problem hiding this comment.
nit: Good to create a method in helpers file for appending metadata like we have for all other headers
| if os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true": | ||
| metadata = list(call_details.metadata or []) | ||
| metadata.append(("x-goog-spanner-enable-afe-server-timing", "true")) | ||
| call_details = call_details._replace(metadata=metadata) |
There was a problem hiding this comment.
Instead of doing call_details._replace which might be a little expensive operation , this can be done with all other metadata.
Eg: We could do this where we are appending prefix metadata which is already called for every operation
| ) | ||
| self.enabled = enabled | ||
| self.gfe_enabled = gfe_enabled | ||
| self.gfe_enabled = True |
There was a problem hiding this comment.
nit: We could remove this
There was a problem hiding this comment.
Addressed removed gfe_enabled completely
| not self.enabled | ||
| or not HAS_OPENTELEMETRY_INSTALLED | ||
| or not getattr(self, "_instrument_afe_latency", None) | ||
| or os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() == "true" |
There was a problem hiding this comment.
nit: Same here, dont fetch from ENV for every call, have a static property and use it.
| header_vals = [] | ||
| for key, val in items: | ||
| key_str = key.decode("utf-8") if isinstance(key, bytes) else str(key) | ||
| if key_str and key_str.lower() in ("server-timing", "server_timing"): |
There was a problem hiding this comment.
We are only capturing "server-timing" in all other clients not "server_timing"
| pass | ||
|
|
||
| if afe_latency is None: | ||
| match = re.search(r"afe(?:t4t7)?;\s*dur=([0-9.]+)", header_val) |
There was a problem hiding this comment.
nit: Optional "t4t7" is not required, it will always be like "/afe; dur=([0-9]+).*/"
a1532ce to
3f91b5f
Compare
Description
This PR enables and properly extracts Spanner Google Front End (GFE) and Application Front End (AFE) latency metrics.
Key Changes:
gfe_enabledtoggle inSpannerMetricsTracerFactory. GFE metrics capture is now always-on whenever OpenTelemetry tracing is enabled.